package com.thiswhyme.mq.kafka.producer.producer;

import com.thiswhyme.mq.kafka.common.consts.TopicNameDefinition;
import com.thiswhyme.mq.kafka.common.pojo.TestMsgDto;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @description: 生产者
 * @author: thiswhyme
 * @date: 2021/4/21
 */
@Slf4j
@AllArgsConstructor
@Component
public class KafkaMsgProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 发送普通消息
     * @param msg
     */
    public void sendCommonMessage(TestMsgDto msg) {
        kafkaTemplate.send(TopicNameDefinition.COMMON_TOPIC, msg.toString());
    }

    /**
     * 发送普通消息2
     * @param msg
     */
    public void sendCommonMessage2(String msg) {
        kafkaTemplate.send(TopicNameDefinition.COMMON_TOPIC2, msg);
    }

    /**
     * 发送可过滤消息
     * @param msg
     */
    public void sendFilterMessage(String msg) {
        kafkaTemplate.send(TopicNameDefinition.FILTER_TOPIC, "kafka-consumer1", msg);
    }

    /**
     * 发送可转发消息
     * @param msg
     */
    public void sendResendMessage(String msg) {
        kafkaTemplate.send(TopicNameDefinition.RESEND_TOPIC, msg);
    }

    /**
     * 发送定时消费消息
     * @param msg
     */
    public void sendTimingMessage(String msg) {
        kafkaTemplate.send(TopicNameDefinition.TIMING_TOPIC, msg);
    }

    /**
     * 发送回调确认消息
     * @param callbackMessage
     */
    public void sendCallbackMessage1(String callbackMessage) {
        kafkaTemplate.send(TopicNameDefinition.CALLBACK_TOPIC, callbackMessage).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            log.info("callback消息【1】发送成功，{}，{}，{}", topic, partition, offset);
        }, failure -> {
            log.info("callback消息【1】发送失败，原因：{}", failure.getMessage(), failure.getCause());
        });
    }

    /**
     * 发送回调确认消息2
     * @param callbackMessage
     */
    public void sendCallbackMessage2(String callbackMessage) {
        kafkaTemplate.send(TopicNameDefinition.CALLBACK_TOPIC, callbackMessage)
                .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("callback消息【2】发送失败，原因：{}", ex.getMessage(), ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("callback消息【2】发送成功，{}，{}，{}", result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(), result.getRecordMetadata().offset());
            }
        });
    }

    /**
     * 发送事务消息
     */
    public void sendTransactionMessage(String msg){
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(TopicNameDefinition.TRANSACTION_TOPIC, msg + 1);
            return true;
        });

        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(TopicNameDefinition.TRANSACTION_TOPIC, msg + 2);
            throw new RuntimeException("fail");
        });
    }
}
